package com.ibm.cps.publisher;

import java.math.BigInteger;
import java.util.HashSet;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.interfaces.AbstractMessagePublisher;
import com.ibm.interfaces.IMessageParser;
import com.ibm.mongo.MongoNoSqlDB;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class MongoPublisher extends AbstractMessagePublisher implements KryoSerializable{
	private static final Logger logger = Logger.getLogger(MongoPublisher.class);
	private static final String CONNECTION_STRING = "connection_string";
	private static final String DATABASE_NAME = "db_name";
	private static final String COLLECTION = "collection";
	
	private String connectionString;
	private String dbName;
	private String collection;
	private ArrayNode buffer = JSONUtility.newArrayNode();
	private MongoNoSqlDB db;
	private int collectionIndex;

	public MongoPublisher(String metadata) throws CPSException {
		super(metadata);
		ObjectNode node = JSONUtility.fromObjectString(metadata);
		this.connectionString = node.get(CONNECTION_STRING).asText();
		this.dbName = node.get(DATABASE_NAME).asText();
		this.collection = node.get(COLLECTION).asText();
		this.db = new MongoNoSqlDB(connectionString, dbName);
		this.collectionIndex = db.getCollection(collection);
	}

	@Override
	public void publishMessage(ParsedDataMessage message,
			IMessageParser parser, boolean isBuffer) throws CPSException {
		String str = new String(parser.restoreFromDataMessage(message));
		ObjectNode node = JSONUtility.fromObjectString(str);
		
		buffer.add(node);
		if (!isBuffer) {
			ObjectNode result = JSONUtility.newObjectNode();
			String id = new BigInteger(new ObjectId().toByteArray()).toString(16);
			result.put("id", id);
			result.set("message", buffer);
			HashSet<String> tagset = new HashSet<String>();
			tagset.add("id");
			tagset.add("message");
			db.upsert(collectionIndex, result, "id", tagset);
			logger.info("insert to mongo :" + result.toString());
			buffer.removeAll();
		}
	}

	@Override
	public MetadataInputSchema getInputSchema() {
		MetadataInputSchema schema = new MetadataInputSchema();
		schema.addField(FieldType.String, CONNECTION_STRING, DATABASE_NAME, COLLECTION);
		return schema;
	}

	@Override
	public void close() {
		try {
			this.db.close();
		} catch (CPSException ex) {
			throw new IllegalStateException("Failed to disconnect db in serializing SeyrenRuleMongoPublisher, DB=" + db, ex);
		}
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(connectionString);
		output.writeString(dbName);
		output.writeString(collection);
		kryo.writeObject(output, buffer);
		try {
			this.db.close();
		} catch (CPSException ex) {
			throw new IllegalStateException("Failed to disconnect db in serializing MongoPublisher, DB=" + db, ex);
		}
	}

	@Override
	public void read(Kryo kryo, Input input) {
		this.connectionString = input.readString();
		this.dbName = input.readString();
		this.collection = input.readString();
		this.buffer = kryo.readObject(input, ArrayNode.class);
		this.db = new MongoNoSqlDB(connectionString, dbName);
		try {
			this.collectionIndex = db.getCollection(collection);
		}catch(CPSException ex){
			throw new IllegalStateException("Failed to connect to db in de-serializing MongoPublisher, DB URI=" + connectionString + ", collection=" + collection, ex);
		}
	}

}
